package com.casic.quality.flink

// com.casic.quality.flink.MyAggAuto

import org.apache.flink.api.java.io.jdbc.JDBCInputFormat
import org.apache.flink.api.scala._

object MyAggAuto extends App {
  args.foreach(arg => System.out.println(arg))
  val env = ExecutionEnvironment.getExecutionEnvironment
  val ds2 = env.createInput(JDBCInputFormat.buildJDBCInputFormat()
    .setDBUrl("jdbc:mysql://10.0.9.187:30435/db_buz_business?characterEncoding=utf-8&autoReconnect=true&&rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai&useSSL=false")
    .setUsername("root")
    .setPassword("r9xh*yH*DcEFQE")
    .setDrivername("com.mysql.cj.jdbc.Driver")
    .setQuery("select * from buz_tools_warn_info_new")
    .finish())
  ds2.print()
  val dataset1 = env.fromCollection(List(("a", 1, 1.1, args(0)), ("b", 2, 2.2, args(0)), ("c", 3, 3.3, args(0)), ("b", 4, 4.4, args(0)), ("c", 5, 5.5, args(0))))
  dataset1.sum(2).max(1).print()
  val num = dataset1.count()
  dataset1.filter(item => item._2 % 2 == 0).sum(3).print()
  dataset1.filter(item => item._2 % 2 != 0).map(item => (item, num)).sum(3).print()
}
